﻿///////////////////////////////////////////////////////////////////////////////
// File:        Server.cc
// Author:      671643387@qq.com
// Description:
///////////////////////////////////////////////////////////////////////////////

#include <boost/bind.hpp>

#include "nio/net/Server.hpp"
using namespace nio;
using namespace nio::net;

Server::Server()
    : state_(READY)
    , connIdGen_(0)
    , connLoad_(0)
    , connTotal_(0)
    , bytsSend_(0)
    , bytsRecv_(0)
    , cnfThreads_(0)
    , cnfThreadFrameTime_(16)
    , cnfServerFrameTime_(16)
    , cnfConnectionHeartbeatTime_(60000)
    , cnfUseSockBuffSet_(false)
    , cnfSndBuffSize_(0x1000)
    , cnfRcvBuffSize_(0x1000)
    , cnfConnectionInputStreamSize_(0x1000)
    , cnfConnectionOutputStreamSize_(0x1000)
    , cnfConnectionUserIO_(false)
#if defined(__LINUX__)
    , cnfEpollET_(true)
#endif
    , connFactory_(new ConnectionFactory())
    , eventLoop_()
    , acceptor_()
    , threads_()
    , connections_(512)
{

}

Server::~Server()
{

}

bool Server::open(std::string addr, int port, int backlog)
{
    if (!acceptor_.open(addr, port, backlog))
        return false;

    if (!eventLoop_.getReactor()->addSource(
        acceptor_.getSocket().getSOCKET(),
        Reactor::RE_RDERR,
        &acceptor_
        ))
    {
        return false;
    }

    acceptor_.setCloseFN(boost::bind(
        &Server::doShutdownAcceptor, this));
    acceptor_.setConnectionAcceptFN(boost::bind(
        &Server::doConnectionAccept, this, _1));

    if (!initThreads())
        return false;

    if (!startThreads())
        return false;

    eventLoop_.setFrameTime(cnfServerFrameTime_);
    eventLoop_.getReactor()->setHeartbeatTime(cnfConnectionHeartbeatTime_);
#if defined(__LINUX__)
    if (cnfEpollET_)
        eventLoop_.getReactor()->setMode(Reactor::MD_ET);
    else
        eventLoop_.getReactor()->setMode(Reactor::MD_LT);
#endif

    return true;
}

void Server::close()
{
    assert(state_ == RUNNING);
    setState(EXITING);
    eventLoop_.queue(boost::bind(&Server::doShutdown, this));
}

void Server::run()
{
    assert(state_ == READY);
    setState(RUNNING);
    eventLoop_.run();
    setState(EXIT);
}

bool Server::initThreads()
{
    if (cnfThreads_ == 0)
        return true;

    for (int x = 0; x < cnfThreads_; ++x)
    {
        ServerThreadPtr thread(new ServerThread());
        if (!thread)
        {
            threads_.clear();
            return false;
        }

        thread->setAddConnectionFN(boost::bind(
            &Server::connectionAdded, this, _1));
        thread->setRemoveConnectionFN(boost::bind(
            &Server::removeConnection, this, _1));
        thread->getLoop().setFrameTime(cnfThreadFrameTime_);
        thread->getLoop().getReactor()->setHeartbeatTime(cnfConnectionHeartbeatTime_);
#if defined(__LINUX__)
        if (cnfEpollET_)
            thread->getLoop().getReactor()->setMode(Reactor::MD_ET);
        else
            thread->getLoop().getReactor()->setMode(Reactor::MD_LT);
#endif
        threads_.push_back(thread);
    }

    return true;
}

bool Server::startThreads()
{
    for (int x = 0; x < cnfThreads_; ++x)
    {
        if (!threads_[x]->start())
        {
            threads_.clear();
            return false;
        }
    }

    return true;
}

void Server::shutdownThreads()
{
    for (int x = 0; x < cnfThreads_; ++x)
    {
        threads_[x]->stop();
        threads_[x]->join();
    }

    eventLoop_.stop();
}

void Server::doConnectionAccept(sock_t sock)
{
    assert(connFactory_);
    ConnectionPtr conn = connFactory_->create(sock,
        cnfConnectionInputStreamSize_, cnfConnectionOutputStreamSize_);
    if (!conn)
    {
        Socket::close(sock);
        return;
    }

    conn->setUserIO(cnfConnectionUserIO_);

    if (cnfUseSockBuffSet_)
    {
        if (!(conn->setSendBufferSize(cnfSndBuffSize_)
            && conn->setRecvBufferSize(cnfRcvBuffSize_)))
        {
            Socket::close(sock);
            return;
        }
    }

    int connId = ++connIdGen_;
    if (connections_.find(connId) != connections_.end())
        return;

    conn->setId(connId);
    connections_[connId] = conn;

    conn->setReadableFN(connReadableFN_);
    conn->setWritableFN(connWritableFN_);
    conn->setErrorFN(connErrorFN_);
    conn->setHeartbeatFN(connHeatbeatFN_);
    conn->setCloseFN(connCloseFN_);

    conn->setSendBytsFN(boost::bind(&Server::byts_Send, this, _1));
    conn->setRecvBytsFN(boost::bind(&Server::byts_Recv, this, _1));

    if (cnfThreads_ == 0)
    {
        if (!eventLoop_.getReactor()->addSource(sock,
            Reactor::RE_ALL, conn.get()))
        {
            connections_.erase(connections_.find(connId));
            return;
        }

        ++connLoad_;
        conn->setShutdownFN(boost::bind(&Server::shutdownConnection, this, _1));
        doConnectionAdded(connections_[connId]);
    }
    else
    {
        ServerThreadArray::size_type freeThreadIdx = 0;
        int minConnInThread = INT_MAX;
        for (int x = 0; x < cnfThreads_; ++x)
        {
            int threadLoad = threads_[x]->getLoad();
            if (threadLoad < minConnInThread)
            {
                freeThreadIdx = ServerThreadArray::size_type(x);
                minConnInThread = threadLoad;
            }
        }

        if (connLoad_ < minConnInThread)
        {
            if (!eventLoop_.getReactor()->addSource(sock,
                Reactor::RE_ALL, conn.get()))
            {
                connections_.erase(connections_.find(connId));
                return;
            }

            ++connLoad_;
            conn->setShutdownFN(boost::bind(&Server::shutdownConnection, this, _1));
            doConnectionAdded(connections_[connId]);
        }
        else
            threads_[freeThreadIdx]->addConnection(connections_[connId]);
    }
}

void Server::connectionAdded(ConnectionPtr& conn)
{
    inc_ConnTotal();
    connectionEstablishedFN_(conn);
}

void Server::doConnectionAdded(ConnectionPtr& conn)
{
    inc_ConnTotal();
    connectionEstablishedFN_(conn);
}

void Server::shutdownConnection(Connection* conn)
{
    eventLoop_.queue(boost::bind(&Server::doShutdownConnection, this, conn));
}

void Server::doShutdownConnection(Connection* conn)
{
    eventLoop_.getReactor()->removeSource(conn->getSocket().getSOCKET());
    --connLoad_;
    doRemoveConnection(conn);
}

void Server::removeConnection(Connection* conn)
{
    eventLoop_.queue(boost::bind(&Server::doRemoveConnection, this, conn));
}

void Server::doRemoveConnection(Connection* conn)
{
    conn->close();
    connections_.erase(connections_.find(conn->getId()));
    dec_ConnTotal();
    if ((state_ == EXITING) && connections_.empty())
        shutdownThreads();
}

void Server::doShutdownAcceptor()
{
    eventLoop_.getReactor()->removeSource(acceptor_.getSocket().getSOCKET());
    acceptor_.shutdown();
}

void Server::doShutdown()
{
    if (acceptor_.getState() == Acceptor::RUNNING)
        doShutdownAcceptor();

    if (connections_.empty())
        shutdownThreads();
    else
    {
        ConnectionMap::iterator it = connections_.begin();
        ConnectionMap::iterator end = connections_.end();
        for (; it != end; ++it)
            it->second->shutdown();
    }
}


